const {
	RSocketConnector
} = require("./@rsocket/core");
const {
	Codec,
	RSocketRequester,
	RSocketResponder
} = require("./@rsocket/messaging");
const {
	RxRequestersFactory,
	RxRespondersFactory,
} = require("./@rsocket/adapter-rxjs");
const {
	TcpClientTransport
} = require("./@rsocket/transport-tcp-client");
// 日志
const Logger = require("../setLog.js");
const {
	firstValueFrom,
	map,
	Observable,
	tap,
	timer,
	interval,
	take,
} = require("rxjs");
class StringCodec {
	mimeType = "text/plain";
	setMimeType(type) {
		thsi.mimeType = type;
	}
	decode(buffer) {
		return buffer.toString();
	}
	encode(entity) {
		return Buffer.from(entity);
	}
}
class ReconnectableRSocket {
	options;
	rsocket;
	activeType;
	stringCodec;
	constructor(options) {
		this.options = options;
		this.activeType = -1;
		this.stringCodec = new StringCodec();
		// this.connect();
	}
	makeConnector(type) {
		let _this = this;
		let mimeTypeList = ['application/json', 'application/octet-stream'];
		return new RSocketConnector({
			setup: {
				dataMimeType: mimeTypeList[type], //application/octet-stream
				keepAlive: 1000000, // avoid sending during test
				lifetime: 100000,
				// metadataMimeType: 'text/plain',
				metadataMimeType: 'message/x.rsocket.routing.v0',
			},
			transport: new TcpClientTransport({
				connectionOptions: {
					host: _this.options.host,
					port: _this.options.port,
				},
			}),
		});
	}
	async connect(type) {
		let _this = this;
		_this.activeType = type;
		const connector = this.makeConnector(type);
		let rsocket = await connector.connect();
		_this.rsocket = RSocketRequester.wrap(rsocket);
	}
	closeSocket() {
		let _this = this;
		if (_this.rsocket) {
			_this.rsocket.close();
			_this.rsocket = null;
		}
	}
	async requestResponse(data, route, type) {
		let _this = this;
		if (!type) {
			type = 0;
		}
		if (!_this.rsocket) {
			Logger.info("连接关闭，重新连接")
			await _this.connect(type);
			// throw new Error('Not Connected yet. Retry later');
		} else if (_this.activeType != type) {
			_this.closeSocket();
			await _this.connect(type);
		}
		console.log("发送");
		console.log(route)
		return new Promise((resolve, reject) => {
			try {
				firstValueFrom(
					_this.rsocket
					.route(route)
					.request(
						RxRequestersFactory.requestResponse(
							data,
							_this.stringCodec,
							_this.stringCodec
						)
					)
					.pipe(tap((res) => {
						console.log("返回");
						console.log("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
						console.log(res);
						resolve(res);
						console.log("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
					}))
				);
			} catch (e) {
				console.log("erro");
				reject(e)
			}
		})

	}
	async requestStream(data, route, type) {
		let _this = this;
		if (!type) {
			type = 0;
		}
		if (!_this.rsocket) {
			Logger.info("连接关闭，重新连接")
			await _this.connect(type);
			// throw new Error('Not Connected yet. Retry later');
		} else if (_this.activeType != type) {
			_this.closeSocket();
			await _this.connect(type);
		}
		return new Promise((resolve, reject) => {
			try {
				_this.rsocket
					.route(route)
					.request(
						RxRequestersFactory.requestStream(
							data,
							_this.stringCodec,
							_this.stringCodec,
							5
						)
					)
					.pipe(
						tap((res) => {
							console.log(res);
							resolve(res);
						}),
						take(25)
					).toPromise();
			} catch (e) {
				console.log("erro");
				reject(e)
			}
		})
	}
	async requestChannel(data, route, type) {
		let _this = this;
		if (!type) {
			type = 0;
		}
		if (!_this.rsocket) {
			Logger.info("连接关闭，重新连接")
			await _this.connect(type);
			// throw new Error('Not Connected yet. Retry later');
		} else if (_this.activeType != type) {
			_this.closeSocket();
			await _this.connect(type);
		}
		return new Promise((resolve, reject) => {
			try {
				_this.rsocket
					.route(route)
					.request(
						RxRequestersFactory.requestChannel(
							interval(1000).pipe(
								map((i) => data),
								tap((res) => {
									console.log("[client] produced")
									console.log(res)
									Logger.info(`[client] produced[data: ${res}]`);
								})
							),
							stringCodec,
							stringCodec,
							5
						)
					)
					.pipe(
						tap((res) => {
							console.log("[client] received")
							console.log(res)
						}),
						take(25)
					)
					.toPromise();
			} catch (e) {
				console.log("erro");
				reject(e)
			}
		})
	}
}
module.exports = ReconnectableRSocket;
